package bigdata.scala.spark.streaming.ycfx.electricstatusqueue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter

import bigdata.scala.spark.streaming.ycfx.model.{EleCurveModel, ElectricMpdel, MyConstants}
import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.HBaseClient
import com.alibaba.fastjson.JSON
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.api.java.JavaInputDStream


/**
 * @author wuzhijiang
 *  获取kafka中电流数据写入hbase中
 */
object ElectricStatusQueue {
  def getTgStatus(args: Array[String]): Unit = {

    val simpleSparkKafka = new SimpleSparkKafka(args)
    simpleSparkKafka.execute(new SKProcessor {
      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        stream.foreachRDD{rdd=>
          if(!rdd.isEmpty()){
            val baseData = rdd.map(kafka => JSON.parseObject(kafka.value()))
              .map((json => {
                val meterId = json.getString("METER_ID")
                val meterAsseton = json.getString("METER_ASSETON")
                val dataDate = json.getString("DATA_DATE")
                val phase = json.getString("PHASE")
                val dotdensity = json.getString("DOTDENSITY")
                val i1  = json.getString("I1")
                val i5  = json.getString("I5")
                val i9  = json.getString("I9")
                val i13 = json.getString("I13")
                val i17 = json.getString("I17")
                val i21 = json.getString("I21")
                val i25 = json.getString("I25")
                val i29 = json.getString("I29")
                val i33 = json.getString("I33")
                val i37 = json.getString("I37")
                val i41 = json.getString("I41")
                val i45 = json.getString("I45")
                val i49 = json.getString("I49")
                val i53 = json.getString("I53")
                val i57 = json.getString("I57")
                val i61 = json.getString("I61")
                val i65 = json.getString("I65")
                val i69 = json.getString("I69")
                val i73 = json.getString("I73")
                val i77 = json.getString("I77")
                val i81 = json.getString("I81")
                val i85 = json.getString("I85")
                val i89 = json.getString("I89")
                val i93 = json.getString("I93")

                ElectricMpdel(meterId,meterAsseton,dataDate,phase,dotdensity,i1,i5,i9,i13,i17,i21,i25,i29,i33,i37,i41,i45,i49,i53,i57,i61,i65,i69,i73,i77,i81,i85,i89,i93)
              }))

          }
        }
      }
    }).start()
  }

  // 将用户电压写入hbase
  def putUserPowerHbase(electricMpdel : ElectricMpdel) = {
    val dataDate:String = LocalDateTime.parse(electricMpdel.dataDate, DateTimeFormatter.ofPattern("yyyyMMdd")).toString // 获取时间
    putUserPowerHbaseBy24Point(dataDate, electricMpdel)

  }

  // 将24个点用户电压写入hbase
  def putUserPowerHbaseBy24Point(dataDate : String, electricMpdel : ElectricMpdel): Unit = {
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 5, electricMpdel.i5, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 9, electricMpdel.i9, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 13, electricMpdel.i13, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 17, electricMpdel.i17, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 21, electricMpdel.i21, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 25, electricMpdel.i25, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 29, electricMpdel.i29, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 33, electricMpdel.i33, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 37, electricMpdel.i37, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 41, electricMpdel.i41, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 45, electricMpdel.i45, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 49, electricMpdel.i49, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 53, electricMpdel.i53, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 57, electricMpdel.i57, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 61, electricMpdel.i61, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 65, electricMpdel.i65, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 69, electricMpdel.i69, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 73, electricMpdel.i73, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 77, electricMpdel.i77, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 81, electricMpdel.i81, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 85, electricMpdel.i85, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 89, electricMpdel.i89, electricMpdel.meterId, electricMpdel.meterAsseton)
    putUserPowerHbaseByPoint(dataDate, electricMpdel.phase, 93, electricMpdel.i93, electricMpdel.meterId, electricMpdel.meterAsseton)

  }

  // 将一个点的数据写入hbase
  def putUserPowerHbaseByPoint(dataDate : String, phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): Unit = {
    // 通过点数生成rowkey
    var rowkey = genRowkey(dataDate, point)
    // 从hbase获取用户电压
    var eleCurveModel = getUserPowerHbase(rowkey)
    // 生成用户电压数据
    genUserPower(eleCurveModel,phase,point,powerValue,meterId,meterAssetno)

    // 将生成新的用户电压写入hbase
    HBaseClient.getInstance().put(MyConstants.tab_user_power, rowkey, eleCurveModel)

  }

  // 生成rowkey,point代表是第几个点 u1是第一个点 以此类推
  def genRowkey(dataDate : String, point : Int) = {
    var hourNumber = (point - 1) * 15 / 60; //十五分钟一个点，求出来的就是第几个小时
    var dataDateTime = ""
    if (hourNumber<10) {
      dataDateTime = "0" + hourNumber.toString
    }else{
      dataDateTime = hourNumber.toString
    }
    dataDateTime
  }

  // 根据rowkey获取hbase中用户电压
  def getUserPowerHbase(rowkey : String) : EleCurveModel = {
    //如果hbase有则取hbase中的数据
    if (HBaseClient.getInstance().exists(MyConstants.tab_user_power, rowkey)) {
      HBaseClient.getInstance().get(MyConstants.tab_user_power, rowkey)
    }
  }

  // 生成电压数据
  def genUserPower(eleCurveModel : EleCurveModel, phase : String, point : Int, powerValue : String, meterId : String, meterAssetno : String): Unit = {
    // 如果不存在用户电压则初始化用户电压
    if (eleCurveModel == null) {

      eleCurveModel.meterId = meterId
      eleCurveModel.meterAssetno = meterAssetno
    }
    phase match {
      case "1" => {
        eleCurveModel.aphase = powerValue
      }
      case "2" => {
        eleCurveModel.bphase = powerValue
      }
      case "3" => {
        eleCurveModel.cphase = powerValue
      }
      case "0" => {
        eleCurveModel.zphase = powerValue
      }
    }
  }
}
